package procedure0;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.HashMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;

public class MR_ResultLimitMapper extends Mapper<Object, Text, Text, Text> {
	
	private HashMap<String, String> amma;
	private String filePath;

	public void setup(Context context) throws IOException {
			
		Configuration jobconf = context.getConfiguration();
		FileSystem fs = FileSystem.get(jobconf);
		filePath = jobconf.get("eigenvaluePath");
		amma = new HashMap<String, String>();
		int i = 0;
		Path f = new Path(filePath + "/Eigenvalue" + i + "/part-r-00000");
		while (fs.exists(f)) {
			
			FSDataInputStream fin = fs.open(f);
			BufferedReader br = new BufferedReader(new InputStreamReader(fin, "UTF-8"));

			String[] sAmma = br.readLine().split("\t", 2);
			amma.put(sAmma[0], sAmma[1]);
					
			br.close();
			i++;
			f = new Path(filePath + "/Eigenvalue" + i + "/part-r-00000");
		} // while
	}

	public void map(Object key, Text value, Context context
			) throws IOException, InterruptedException {
		
		String s = value.toString().split("\t")[0];
		String[] strArray = s.split(",", -1);	// limit value is -1 means split the most segments 
		String[] resultList = new String[5];
		resultList[0] = String.valueOf(strArray[0]);
		resultList[1] = String.valueOf(strArray[1]);
		resultList[2] = String.valueOf(Double.valueOf(strArray[2]));
		resultList[3] = String.valueOf(Double.valueOf(strArray[3]));
		resultList[4] = String.valueOf(Double.valueOf(strArray[4]));

		s = resultList[0];
		for (int i=1; i<resultList.length; i++) {
			s += ",";
			s += resultList[i];
		}
		context.write(new Text(s), new Text(""));
	}
}

